package test

import (
	"fmt"
	"github.com/IBM/sarama"
	"src/config"
	"src/connection"
	"src/global"
	"src/model/ProtoModel"
	"src/service"
	"sync"
	"time"
	// 	"unsafe"
	// 	"strconv"
)

var b int64 = 0

var wg sync.WaitGroup

// 回掉函数
func Callbake(s *sarama.PartitionConsumer, v *sarama.ConsumerMessage) {
	if s == nil {

	}
	fmt.Println(string(v.Value))
	return
}

var idx int64

// 订阅消息 格式封包
func packetMessage() *ProtoModel.MessagesPubProto {
	//var mes []ProtoModel.MessageProto
	//mes = append(mes, ProtoModel.MessageProto{Action: "XdfdsfdsdfdsX", Image: "1"})
	//mes = append(mes, ProtoModel.MessageProto{Action: "XXxxxxxxxxxxfeeXX", Image: "1"})
	//bug := service.Sub{Sid: "xxfd", Uid: "1084", Cid: idx}
	//bug.S.Lock()
	//bug := bytes.Buffer{}
	var mes2 ProtoModel.MessagesPubProto
	mes2.Did = "h5"
	mes2.Msg = append(mes2.Msg, &ProtoModel.MessageProto{Action: "XdfdsfdsdfdsX", Image: "1"})
	mes2.Msg = append(mes2.Msg, &ProtoModel.MessageProto{Action: "156654899984X", Image: "2"})
	//bug.WriteByte(1)
	//bug.WriteByte(2)

	//bug.Message(&mes2)
	//idx++
	return &mes2
}
func Push(topic string, key string) {
	for {
		idx++
		sub1 := service.Sub{Sid: "xxfd", Uid: "1084", Cid: idx}
		sub1.Act = ProtoModel.SendActProto_SENDUID
		sub1.Type = ProtoModel.DataTypeProto_MESSAGE
		sub1.Message(packetMessage())
		err := sub1.SendKafka()
		if err != nil {
			return
		}

		//time.Sleep(1 * time.Second)
	}
}
func main() {
	//加载配置
	config.Init()
	//全局变量
	global.Init()
	//connection.InIt() //初始化连接redis
	//connection.NewRedisClient()
	fmt.Println("a=", "start_____________________") // a=10
	//ch2 := make(chan []byte)
	go (&connection.PubKafkaConsumer{}).NewKafkaConsumer().Subscribe("chat", Callbake)
	//connection.KafkaConsumerGroup()
	Push("chat", "mes")
	time.Sleep(3 * time.Second)
	return
	// 	fmt.Println("a=", userInfos2==nil)  // a=10

}
